package main.ch12.java;

import java.io.IOException;

import main.common.java.NcdcRecordParser;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.BatchUpdate;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.NullOutputFormat;
import org.apache.hadoop.util.*;

public class HBaseTemperatureImporter extends Configured implements Tool {

	// Inner-class for map
	static class HBaseTemperatureMapper<K, V> extends MapReduceBase implements
			Mapper<LongWritable, Text, K, V> {
		private NcdcRecordParser parser = new NcdcRecordParser();
		private HTable table;

		public void map(LongWritable key, Text value,
				OutputCollector<K, V> output, Reporter reporter)
				throws IOException {
			parser.parse(value.toString());
			if (parser.isValidTemperature()) {
				byte[] rowKey = RowKeyConverter.makeObservationRowKey(parser
						.getStationId(), parser.getObservationDate().getTime());
				BatchUpdate bu = new BatchUpdate(rowKey);
				bu.put("data:airtemp",
						Bytes.toBytes(parser.getAirTemperature()));
				table.commit(bu);
			}
		}

		public void configure(JobConf jc) {
			super.configure(jc);
			// Create the HBase table client once up-front and keep it around
			// rather than create on each map invocation.
			try {
				this.table = new HTable(new HBaseConfiguration(jc),
						"observations");
			} catch (IOException e) {
				throw new RuntimeException("Failed HTable construction", e);
			}
		}
	}

	public int run(String[] args) throws IOException {
		if (args.length != 1) {
			System.err.println("Usage: HBaseTemperatureImporter <input>");
			return -1;
		}
		JobConf jc = new JobConf(getConf(), getClass());
		FileInputFormat.addInputPath(jc, new Path(args[0]));
		jc.setMapperClass(HBaseTemperatureMapper.class);
		jc.setNumReduceTasks(0);
		jc.setOutputFormat(NullOutputFormat.class);
		JobClient.runJob(jc);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		String [] arg = {"job-in01"};
		int exitCode = ToolRunner.run(new HBaseConfiguration(),
				new HBaseTemperatureImporter(), arg);
		System.exit(exitCode);
	}
}